package cn.dglydrpy.study.common.kafka;

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerEndpointRegistry;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.stereotype.Component;

/**
 * 设置定时任务 对kafka消费者监听器做启停控制
 */
@Component
@EnableScheduling
@Slf4j
public class KafkaTaskListener{

    @Autowired
    private KafkaListenerEndpointRegistry registry;

    @Autowired
    private ConsumerFactory consumerFactory;

    /**
     * 禁止自动启动
     * @return
     */
    @Bean
    public ConcurrentKafkaListenerContainerFactory delayContainerFactory() {
        ConcurrentKafkaListenerContainerFactory container = new ConcurrentKafkaListenerContainerFactory();
        container.setConsumerFactory(consumerFactory);
        container.setAutoStartup(false);
        return container;
    }

    @KafkaListener(id = "durable", topics = "topic.quick.durable",containerFactory = "delayContainerFactory")
    public void durableListener(String data) {
        //这里做数据持久化的操作
//        log.info("topic.quick.durable receive : " + data);
    }


    //定时器，每天凌晨0点开启监听
//    @Scheduled(cron = "0 0 0 * * ?")
//    public void startListener() {
//        log.info("开启监听");
//        //判断监听容器是否启动，未启动则将其启动
//        if (!registry.getListenerContainer("durable").isRunning()) {
//            registry.getListenerContainer("durable").start();
//        }
//        registry.getListenerContainer("durable").resume();
//    }

    //定时器，每天早上10点关闭监听
//    @Scheduled(cron = "0 0 10 * * ?")
//    public void shutDownListener() {
//        log.info("关闭监听");
//        registry.getListenerContainer("durable").pause();
//    }
}
